package com.yunhe.abnormal.websocket;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.common.constant.AbnormalConsts;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.constant.SystemConsts;
import com.yunhe.abnormal.consts.AlarmConsts;
import com.yunhe.common.model.data.AlarmEventDTO;
import com.yunhe.common.model.msg.AlarmEventMsgDTO;
import com.yunhe.abnormal.service.base.AlarmNotifyConfigService;
import com.yunhe.common.util.JsonUtil;
import com.yunhe.common.util.StringUtil;
import com.yunhe.common.websocket.WebsocketRealtimeHandler;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 实时告警推送WebSocket处理类
 * @author 无
 */
public class AlarmNotifyHandler extends WebsocketRealtimeHandler {
    private static final Logger logger = LoggerFactory.getLogger(AlarmNotifyHandler.class);
    private ObjectMapper mapper = JsonUtil.getObjectMapper();
    @Autowired
    private AlarmNotifyConfigService alarmNotifyConfigService;
    /**
     * sessionId和用户ID的映射
     */
    private Map<String, Long> userIdMap = new ConcurrentHashMap<>();
    /**
     * sessionId和单位ID的映射
     */
    private Map<String, Long> firmIdMap = new ConcurrentHashMap<>();

    @Override
    public String getPointField() {
        return "stationIds";
    }

    @Override
    @SuppressWarnings("unchecked")
    public void beforeSend(WebSocketSession session, Object content) {
        Long userId = userIdMap.get(session.getId());
        Long firmId = firmIdMap.get(session.getId());
        Map<String, List<AlarmEventDTO>> result = (Map<String, List<AlarmEventDTO>>)content;
        Iterator<String> iterator = result.keySet().iterator();
        while(iterator.hasNext()) {
            String point = iterator.next();
            List<AlarmEventDTO> alarmEvents = result.get(point);
            alarmEvents.removeIf(alarmEvent -> !alarmNotifyConfigService.needNotifyByUser(alarmEvent, AlarmConsts.AlarmNotifyType.POPUP, userId, firmId));
            if (alarmEvents.size() == 0) {
                iterator.remove();
                result.remove(point);
            }
        }
    }

    @Override
    public void afterEstablished(WebSocketSession session) {
        String userIdStr = this.getUrlParam(session, SystemConsts.HEADER.USER_ID.value());
        if (!StringUtil.isEmpty(userIdStr)) {
            userIdMap.put(session.getId(), Long.parseLong(userIdStr));
        }
        String firmIdStr = this.getUrlParam(session, SystemConsts.HEADER.FIRM_ID.value());
        if (!StringUtil.isEmpty(firmIdStr)) {
            firmIdMap.put(session.getId(), Long.parseLong(firmIdStr));
        }
    }

    @Override
    public void afterClosed(WebSocketSession session, CloseStatus status) {
        userIdMap.remove(session.getId());
        firmIdMap.remove(session.getId());
    }

    @Override
    public void handleMessage(WebSocketSession session, TextMessage message, List<String> points) {

    }

    /**
     * 监听告警新增消息
     * @param record 消息内容：主要是电站ID
     */
    @KafkaListener(topics = KafkaConsts.Topic.AlarmEventRefresh, groupId = KafkaConsts.Group.RandomGroupId)
    public void alarmNotifyListener(ConsumerRecord<?, String> record) {
        try {
            String value = record.value();
            if (!StringUtil.isEmpty(value)) {
                AlarmEventMsgDTO msg = mapper.readValue(value, AlarmEventMsgDTO.class);
                String action = msg.getAction();
                if (SystemConsts.ACTION.ADD.equals(action)) {
                    List<AlarmEventDTO> alarmEvents = msg.getAlarmEvents();
                    Map<String, List<AlarmEventDTO>> result = new HashMap<>();
                    List<String> points = new ArrayList<>();
                    for (AlarmEventDTO alarmEvent : alarmEvents) {
                        Long stationId = alarmEvent.getStationId();
                        if (stationId != null) {
                            String point = String.valueOf(stationId);
                            List<AlarmEventDTO> onResult = result.computeIfAbsent(point, v -> new ArrayList<>());
                            onResult.add(alarmEvent);
                            points.add(point);
                        } else {
                            logger.error("告警通知ID:{}未推送，因为电站ID为空", alarmEvent.getId());
                        }
                    }
                    sendMessage(result, null, points);
                }
            }
        } catch (Exception e) {
            logger.error("新增告警实时推送报错，{}，{}", record.value(), e);
        }
    }
}
